//package com.pdd.rocketmq.config;
//
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.*;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//
//import java.io.UnsupportedEncodingException;
//import java.util.List;
//
///**
// * @author:liyangpeng
// * @date:2020/8/24 11:20
// */
//@Configuration
//@Slf4j
//public class RocketMQConfig {
//
//    @Autowired
//    private RocketMQYmlConfig ymlConfig;
//
//
//
//    @Bean
//    public DefaultMQProducer defaultMQProducer(){
//        DefaultMQProducer defaultMQProducer=new DefaultMQProducer(ymlConfig.getGroupName());
//        defaultMQProducer.setNamesrvAddr(ymlConfig.getNode());
//        try {
//            defaultMQProducer.start();
//        } catch (MQClientException e) {
//            log.info("rocket消息提供开启失败:{}",e);
//        }
//        return defaultMQProducer;
//    }
//
//    @Bean
//    public DefaultMQPushConsumer defaultMQPushConsumer(){
//        DefaultMQPushConsumer defaultMQPushConsumer=new DefaultMQPushConsumer(ymlConfig.getGroupName());
//        defaultMQPushConsumer.setNamesrvAddr(ymlConfig.getNode());
//        try {
//            defaultMQPushConsumer.subscribe("pdd-topic-1","*");
//            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
//                @Override
//                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//                    list.stream().forEach(m->{
//                        try {
//                            log.info("收到消息,messageID:{},message:{}",m.getMsgId(),new String(m.getBody(),"UTF-8"));
//                        } catch (UnsupportedEncodingException e) {
//                            log.info("消息解码失败:{}",e);
//                        }
//                    });
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//            });
//            defaultMQPushConsumer.start();
//        } catch (MQClientException e) {
//            log.info("rocket消息消费开启失败:{}",e);
//        }
//        return defaultMQPushConsumer;
//    }
//}